package com.jvm123

import groovy.sql.GroovyRowResult
import groovy.sql.Sql
import io.reactivex.rxjava3.core.Observer
import io.reactivex.rxjava3.disposables.Disposable

import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors

/**
 * @author yawn
 */
class ExecutorObserver implements Observer<Collection<String>> {


    ExecutorService executor = Executors.newFixedThreadPool(6)

    Sql sql = Sql.newInstance(Constant.URL, Constant.USER, Constant.PWD, Constant.DRIVER)

    List<GroovyRowResult> relationList = sql.rows("SELECT * FROM goods_relation")

    {
        sql.execute("update goods_relation set score = 0")
    }

    @Override
    void onSubscribe(Disposable disposable) {
        System.out.println("--- onSubscribe ---")
//        disposable.dispose()
    }
    @Override
    void onNext(Collection<String> collection) {
        executor.execute(new Runnable() {
            @Override
            void run() {
                for (r in relationList) {
                    String id1 = r.get("goods_id1")
                    String id2 = r.get("goods_id2")
                    if (collection.containsAll([id1, id2])) {
                        sql.execute("update goods_relation set score = score + 1 where (goods_id1 = ? and goods_id2 = ?) or (goods_id1 = ? and goods_id2 = ?)", [id1, id2, id2, id1])
                    }
                }
            }
        })
    }
    @Override
    void onError(Throwable throwable) {
        System.out.println("--- onError ---")
    }
    @Override
    void onComplete() {
        System.out.println("--- onComplete ---")
        executor.shutdown()
    }
}